package com.newhope.srm.policy.service.report.conponent;

import com.newhope.srm.policy.common.exception.NoListenerException;
import com.newhope.srm.policy.entity.po.market.MarketIndexData;
import com.newhope.srm.policy.service.report.event.EventData;
import com.newhope.srm.policy.service.report.event.IEventListener;
import lombok.extern.slf4j.Slf4j;
import org.greenrobot.eventbus.Subscribe;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

@Component
@Slf4j
public class EventDispatcher {

    @Resource
    private ApplicationContext applicationContext;


    private static Logger logger = LoggerFactory.getLogger(EventDispatcher.class);

    private ConcurrentHashMap<String, ConcurrentLinkedQueue<IEventListener>> listenerMaps = new ConcurrentHashMap<>();

    @PostConstruct
    private void init() {
        Map<String, IEventListener> beans = applicationContext.getBeansOfType(IEventListener.class);
        beans.entrySet().forEach(item -> {
            registerListeners(item.getValue());
        });
    }

    @Subscribe
    public void onMessageEvent(List<MarketIndexData> event) {
        for (MarketIndexData marketIndexData : event) {
            ConcurrentLinkedQueue<IEventListener> eventListeners = listenerMaps.get(marketIndexData.getIndexCode());
            if (eventListeners != null && !eventListeners.isEmpty()) {
                Iterator<IEventListener> iterator = eventListeners.iterator();
                while (iterator.hasNext()) {
                    IEventListener next = iterator.next();
                    try {
                        CompletableFuture.runAsync(() -> {
                                next.onEvent(marketIndexData);
                        });
                    } catch (Exception e) {
                        logger.error(e.getMessage(), e);
                    }

                }
            }else{
                throw new NoListenerException("请检查编码"+marketIndexData.getIndexCode()+"没有相应的监视器");
            }
        }

    }


    private void registerListeners(IEventListener eventListener) {

        String indexCode = eventListener.getIndexCode();
        ConcurrentLinkedQueue<IEventListener> iEventListeners = listenerMaps.get(indexCode);
        if (iEventListeners == null) {
            iEventListeners = new ConcurrentLinkedQueue<>();
            listenerMaps.put(indexCode, iEventListeners);
        }

        iEventListeners.add(eventListener);
    }

    @PreDestroy
    public void unregisterListeners() {

        Iterator<Map.Entry<String, ConcurrentLinkedQueue<IEventListener>>> iterator = listenerMaps.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, ConcurrentLinkedQueue<IEventListener>> next = iterator.next();
            ConcurrentLinkedQueue<IEventListener> value = next.getValue();
            if (value != null) {
                value.clear();
            }
        }

        listenerMaps.clear();
    }
}
